package org.aurora.core.server.scheduler;

import org.apache.commons.cli.Option;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.aurora.common.delay.queue.DelayExecuteTaskQueue;
import org.aurora.common.thread.ServiceThread;
import org.aurora.common.util.CommandlineUtil;
import org.aurora.common.util.NetworkUtil;
import org.aurora.common.util.SystemTimeUtil;
import org.aurora.common.util.ThreadSleepTimer;
import org.aurora.core.biz.BizBeanAcquirer;
import org.aurora.core.biz.SchedulerSpringApplication;
import org.aurora.core.biz.service.*;
import org.aurora.core.common.SchedulerConstant;
import org.aurora.core.common.exception.ExpressionParseException;
import org.aurora.core.server.scheduler.processor.ClientReportTaskProcessor;
import org.aurora.core.server.scheduler.task.TaskWrapper;
import org.aurora.core.server.scheduler.task.TaskWrapperFactory;
import org.aurora.model.entity.*;
import org.aurora.model.metadata.SchedulerMetaData;
import org.aurora.model.request.AssignHashSlotsRequest;
import org.aurora.model.request.ExecuteTaskRequest;
import org.aurora.model.request.ReportTaskRequest;
import org.aurora.model.response.AssignHashSlotsResponse;
import org.aurora.model.response.CommonResponse;
import org.aurora.model.type.RequestCommandType;
import org.aurora.model.type.TaskState;
import org.aurora.model.type.TaskTriggerType;
import org.aurora.remote.RemoteProcessor;
import org.aurora.remote.RemoteServer;
import org.aurora.remote.netty.*;
import org.aurora.remote.protocol.RemoteResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class DefaultSchedulerImpl implements Scheduler {

    private static final Logger log = LoggerFactory.getLogger(DefaultSchedulerImpl.class);

    // 调度器重平衡锁
    private final Object GLOBAL_BALANCE_LOCK = new Object();

    private DelayExecuteTaskQueue<DelaySchedulerTaskWrapper> slowSchedulerDelayTaskQueue;

    // todo 设置线程数量
    private ExecutorService slowSchedulerQueueExecutor = Executors.newFixedThreadPool(5);

    private ExecutorService fastSchedulerQueueExecutor = Executors.newFixedThreadPool(10);

    private DelayExecuteTaskQueue<DelaySchedulerTaskWrapper> fastSchedulerDelayTaskQueue;

    // 接收请求
    private NettyRemoteServer schedulerRemoteServer;

    // 连接到主调度器的连接
    private MainSchedulerConnector mainSchedulerConnector;

    private volatile SchedulerConfig schedulerConfig;

    // 调度器元数据
    private volatile SchedulerMetaData schedulerMetaData;

    /**
     * task info start
     **/
    private Map<String/*appcode*/, ConcurrentMap<String/*clientIp+:+clientPort*/, List<String>/*taskNames*/>> clientInfoMap = new HashMap<>();

    private ConcurrentHashMap<String/*taskName*/, TaskToClientAddressWrapper> taskNameToClientAddrMap = new ConcurrentHashMap<>();

    private volatile Map<Long, TaskWrapper> taskWrapperMap = new HashMap<>();
    /** task info end **/

    /** main scheduler client start **/

    /** main scheduler client end **/

    /**
     * main scheduler server start
     **/
    private DefaultInnerMainScheduler defaultInnerMainScheduler;

    private final AtomicBoolean isMainScheduler = new AtomicBoolean(false);
    /** main scheduler server end **/

    /**
     * biz bean start
     **/
    private BizBeanAcquirer bizBeanAcquirer;

    private IMainSchedulerService mainSchedulerService;

    private ITaskService taskService;

    private IClientAppService clientAppService;

    private ISchedulerService schedulerService;

    private ISlotAssignService slotAssignService;

    /**
     * biz bean end
     **/

    public DefaultSchedulerImpl(String[] args) throws Exception {
        this.schedulerConfig = this.loadConfig(args);
        // 业务 bean 获取器，当前容器为 spring
        this.bizBeanAcquirer = SchedulerSpringApplication.getInstance();
        // 调度器延迟队列，把任务放到延迟队列中等待到期后取出
        this.slowSchedulerDelayTaskQueue = new DelayExecuteTaskQueue<>(
                "slowSchedulerDelayTaskQueueThread",
                delayTaskWrapper -> {
                    this.slowSchedulerQueueExecutor.execute(() -> {
                        ScheduleTask schedulerTask = delayTaskWrapper.getSchedulerTask();
                        Task task = schedulerTask.getTask();
                        Long taskId = task.getId();
                        Date expectedSchedulerTime = schedulerTask.getExpectedSchedulerTime();
                        TaskWrapper taskWrapper = taskWrapperMap.get(taskId);

                        long currentTimeMillis = 0;

                        do {
                            // todo 写入数据库，考虑异常情况

                            currentTimeMillis = SystemTimeUtil.currentTimeMillis();
                            // 放入到 fast 队列
                            this.fastSchedulerDelayTaskQueue.offer(new DelaySchedulerTaskWrapper(expectedSchedulerTime.getTime() - currentTimeMillis, schedulerTask));

                            schedulerTask = taskWrapper.getNextSchedulerTask(TaskTriggerType.MACHINE, expectedSchedulerTime);
                            expectedSchedulerTime = schedulerTask.getExpectedSchedulerTime();
                        } while (expectedSchedulerTime.getTime() - SystemTimeUtil.currentTimeMillis() <= schedulerConfig.getFastSchedulerQueueAdvanceTime());

                        this.slowSchedulerDelayTaskQueue.offer(new DelaySchedulerTaskWrapper(expectedSchedulerTime.getTime() - currentTimeMillis, schedulerTask));
                    });

                });
        this.fastSchedulerDelayTaskQueue = new DelayExecuteTaskQueue<>(
                "fastSchedulerDelayTaskQueueThread",
                delayTask -> {
                    fastSchedulerQueueExecutor.execute(() -> {
                        doExecuteScheduleTask(delayTask.getSchedulerTask());
                    });
                });

        this.defaultInnerMainScheduler = new DefaultInnerMainScheduler();

        this.schedulerRemoteServer = new NettyRemoteServer(schedulerConfig.getSchedulerPort());

        this.mainSchedulerConnector = new DefaultMainSchedulerConnector(this.schedulerConfig, this::tryToBeMainScheduler);

        // 注册处理器
        this.registerProcessor();
    }

    /**
     * 加载调度器配置
     *
     * @param args
     * @return
     * @throws Exception
     */
    private SchedulerConfig loadConfig(String[] args) throws Exception {
        try {
            return CommandlineUtil.parseCommandLineConfig(args, new SchedulerConfig(), new Option("c", "config", true, "config properties file"));
        } catch (Exception e) {
            log.error("scheduler load config error", e);
            throw e;
        }
    }

    /**
     * 注册处理器
     */
    private void registerProcessor() {
        this.schedulerRemoteServer.registerProcessor(RequestCommandType.CLIENT_REPORT_TASK.getCode(), new ClientReportTaskProcessor(this));
    }

    private void fillBizBean() {
        this.mainSchedulerService = this.bizBeanAcquirer.getBean(IMainSchedulerService.class);
        this.taskService = this.bizBeanAcquirer.getBean(ITaskService.class);
        this.schedulerService = this.bizBeanAcquirer.getBean(ISchedulerService.class);
        this.clientAppService = this.bizBeanAcquirer.getBean(IClientAppService.class);
        this.slotAssignService = this.bizBeanAcquirer.getBean(ISlotAssignService.class);
    }

    @Override
    public void start() throws InterruptedException {
        // 业务 bean 获取器，当前容器为 spring
        this.bizBeanAcquirer.start(this.schedulerConfig.getBizConfigLoadUrl());
        this.fillBizBean();

        // 启动延迟队列进行调度
        this.fastSchedulerDelayTaskQueue.start();
        this.slowSchedulerDelayTaskQueue.start();

        this.defaultInnerMainScheduler.start();
        // 新增处理器

        // 创建 netty server
        this.schedulerRemoteServer.start();

        // 进行选主 和 加载任务
        this.tryToBeMainScheduler();

        // 执行任务
        this.executeTask();
    }


    @Override
    public void shutdown() {
        this.slowSchedulerDelayTaskQueue.stop();
        this.fastSchedulerDelayTaskQueue.stop();
        this.schedulerRemoteServer.shutdown();
        this.defaultInnerMainScheduler.shutdown();
    }

    @Override
    public void executeTask() {
        // 加全局锁
        synchronized (GLOBAL_BALANCE_LOCK) {
            // 做状态清理
            // 清理任务？清理调度延迟队列
            // todo 清除当前调度任务

            // 是主调度器并且不进行工作
            if (this.isMainScheduler.get() && !this.schedulerConfig.isSelfMainSchedulerEnabledWork()) {
                log.info("is main scheduler and no work");
                return;
            }

            List<Task> tasks = this.taskNameToClientAddrMap.values().stream().map(TaskToClientAddressWrapper::getTask).collect(Collectors.toList());

            Map<Long, TaskWrapper> taskWrapperMap = tasks.stream()
                    .map(TaskWrapperFactory::createScheduleTask)
                    .collect(
                            Collectors.toMap(
                                    taskWrapper -> taskWrapper.getTask().getId(),
                                    Function.identity()
                            )
                    );

            this.taskWrapperMap = taskWrapperMap;

            for (TaskWrapper taskWrapper : taskWrapperMap.values()) {
                Task task = taskWrapper.getTask();
                int taskState = task.getState();

                if (!Objects.equals(taskState, TaskState.RUNNING.getCode())) {
                    continue;
                }

                ScheduleTask scheduleTask = taskWrapper.getNextSchedulerTask(TaskTriggerType.MACHINE, new Date());
                long currentTimeMillis = SystemTimeUtil.currentTimeMillis();
                Date expectedSchedulerTime = scheduleTask.getExpectedSchedulerTime();
                this.slowSchedulerDelayTaskQueue.offer(new DelaySchedulerTaskWrapper(expectedSchedulerTime.getTime() - currentTimeMillis, scheduleTask));
            }
        }
    }

    @Override
    public void registerClientTask(String clientIp, int clientPort, String clientAppcode, List<String> taskNames) {
        for (String taskName : taskNames) {
            this.taskNameToClientAddrMap.get(taskName).addClientAddress(NetworkUtil.splicingAddr(clientIp, clientPort));
        }

        // todo 客户端与任务绑定信息写入到数据库中
    }

    @Override
    public void loadTask() {
        List<Integer> schedulerHashSlots = this.schedulerMetaData.getSlots();

        if (CollectionUtils.isNotEmpty(schedulerHashSlots)) {
            // 加载客户端信息
            List<ClientAppInfo> clientAppInfosByHashSlots = this.clientAppService.getClientAppInfosByHashSlots(schedulerHashSlots);

            Map<String/*appcode*/, ConcurrentMap<String/*clientIp+:+clientPort*/, List<String>/*taskNames*/>> clientInfoMap = new HashMap<>();
            ConcurrentHashMap<String/*taskName*/, TaskToClientAddressWrapper> taskNameToClientAddrMap = new ConcurrentHashMap<>();

            if (CollectionUtils.isNotEmpty(clientAppInfosByHashSlots)) {
                for (ClientAppInfo clientAppInfo : clientAppInfosByHashSlots) {
                    String appcode = clientAppInfo.getAppcode();
                    // 加载客户端对应任务
                    // todo 考虑异常情况该怎么办
                    List<Task> tasks = this.taskService.queryTasksByAppcode(clientAppInfo.getAppcode());

                    clientInfoMap.put(appcode, new ConcurrentHashMap<>());

                    if (CollectionUtils.isEmpty(tasks)) {
                        continue;
                    }

                    for (Task task : tasks) {
                        taskNameToClientAddrMap.put(task.getName(), new TaskToClientAddressWrapper(task, new ConcurrentSkipListSet<>()));
                    }
                }
            }

            this.clientInfoMap = clientInfoMap;
            this.taskNameToClientAddrMap = taskNameToClientAddrMap;
        } else {
            // 清空状态
            this.clientInfoMap = new HashMap<>();
            this.taskNameToClientAddrMap = new ConcurrentHashMap<>();
        }
    }

    /**
     * @param scheduleTask
     */
    private void doExecuteScheduleTask(ScheduleTask scheduleTask) {
        Task task = scheduleTask.getTask();

        String taskName = task.getName();
        String appcode = task.getAppcode();

        String scheduleTaskUniqueId = scheduleTask.getUniqueId();

        TaskToClientAddressWrapper taskToClientAddressWrapper = this.taskNameToClientAddrMap.get(task.getName());
        Set<String> clientAddress = null;

        if (taskToClientAddressWrapper == null) {
            log.warn("execute schedule task send request but task not exist, " +
                            "taskName = {}, appcode = {}, schedule task uniqueId = {}",
                    taskName, appcode, scheduleTaskUniqueId);
            return;
        }

        if (CollectionUtils.isEmpty(clientAddress = taskToClientAddressWrapper.getClientAddress())) {
            log.warn("execute schedule task send request but client no node, " +
                            "taskName = {}, appcode = {}, schedule task uniqueId = {}",
                    taskName, appcode, scheduleTaskUniqueId);
            return;
        }

        String clientAddr = null;

        // todo 轮询，后面考虑做负载均衡
        while (true) {
            try {
                clientAddr = clientAddress.stream().findFirst().orElse(null);

                if (StringUtils.isBlank(clientAddr)) {
                    log.warn("execute schedule task send request but response is null, " +
                                    "taskName = {}, appcode = {}, client addr = {}, schedule task uniqueId = {}",
                            taskName, appcode, clientAddr, scheduleTaskUniqueId);
                    // todo 将信息写入到数据库进行报警
                    break;
                }

                scheduleTask.setActualSchedulerTime(new Date());
                CommonResponse response = this.schedulerRemoteServer.send(clientAddr, new ExecuteTaskRequest(task, scheduleTask), 1000, CommonResponse.class);

                if (response == null) {
                    log.warn("execute schedule task send request but response is null, " +
                            "taskName = {}, appcode = {}, client addr = {}, schedule task uniqueId = {}",
                            taskName, appcode, clientAddr, scheduleTaskUniqueId);
                    this.schedulerRemoteServer.disConnect(clientAddr);
                    clientAddress.remove(clientAddr);
                    continue;
                }

                if (!response.isSuccess()) {
                    log.warn("execute schedule task send request but response is no success, " +
                            "taskName = {}, appcode = {}, client addr = {}, schedule task uniqueId = {}",
                            taskName, appcode, clientAddr, scheduleTaskUniqueId);
                    this.schedulerRemoteServer.disConnect(clientAddr);
                    clientAddress.remove(clientAddr);
                    continue;
                }

                break;
            } catch (Exception e) {
                log.error("execute schedule task send request error, " +
                        "taskName = {}, appcode = {}, client addr = {}, schedule task uniqueId = {}",
                        taskName, appcode, clientAddr, scheduleTaskUniqueId, e);
                this.schedulerRemoteServer.disConnect(clientAddr);
                clientAddress.remove(clientAddr);
            }
        }
    }

    /**
     * 尝试成为主调度器
     */
    private void tryToBeMainScheduler() {
        synchronized (GLOBAL_BALANCE_LOCK) {
            final String ip = NetworkUtil.getIpAddress();
            final int selfMainSchedulerPort = schedulerConfig.getSelfMainSchedulerPort();

            SchedulerInfo schedulerInfo = new SchedulerInfo(ip, selfMainSchedulerPort);
            final String curMainSchedulerAddr = schedulerInfo.getAddress();

            for (;;) {
                try {
                    // 获取数据库主调度器地址
                    MainSchedulerLockInfo mainSchedulerLockInfo = mainSchedulerService.queryMainSchedulerLockInfo();
                    String preMainSchedulerAddr = mainSchedulerLockInfo.getPreMainSchedulerAddr();
                    String preMainSchedulerIp = mainSchedulerLockInfo.getPreMainSchedulerIp();
                    int preMainSchedulerPort = mainSchedulerLockInfo.getPreMainSchedulerPort();

                    if (Objects.equals(preMainSchedulerAddr, curMainSchedulerAddr)) {
                        this.isMainScheduler.set(true);
                        this.schedulerConfig.setMainSchedulerIp(ip);
                        this.schedulerConfig.setMainSchedulerPort(selfMainSchedulerPort);
                        this.defaultInnerMainScheduler.assignHashSlot();
                        break;
                    } else {
                        try {
                            // 尝试建立连接
                            this.mainSchedulerConnector.connect(preMainSchedulerIp, preMainSchedulerPort);
                            // 建立连接成功
                            this.isMainScheduler.set(false);
                            this.schedulerConfig.setMainSchedulerIp(preMainSchedulerIp);
                            this.schedulerConfig.setMainSchedulerPort(preMainSchedulerPort);

                            // 等待 hash 槽分配
                            this.mainSchedulerConnector.awaitHashSlotAssign();
                            break;
                        } catch (Exception e) {
                            // 建立连接失败，尝试将当前节点写为主节点
                            log.warn("Failed to connect to main scheduler", e);
                            mainSchedulerLockInfo.setCurMainSchedulerIp(schedulerInfo.getIp());
                            mainSchedulerLockInfo.setCurMainSchedulerPort(schedulerInfo.getPort());
                            mainSchedulerLockInfo.setCurMainSchedulerStartTime(new Date());

                            this.mainSchedulerService.tryToBeMainScheduler(mainSchedulerLockInfo);
                        }
                    }
                } catch (Exception e) {
                    log.error("try to be main scheduler error", e);
                }
            }

            // 加载任务信息
            this.loadTask();
        }
    }

    private class DefaultInnerMainScheduler implements MainScheduler, NettyEventSubscriber {
        private final Logger log = LoggerFactory.getLogger(DefaultInnerMainScheduler.class);

        private final NettyRemoteServer mainSchedulerRemoteServer;

        private final BlockingQueue<NettyEvent> nettyEventQueue = new LinkedBlockingQueue<>();

        // todo 改为配置化
        private final ThreadSleepTimer reAssignHashSleepTimer = new ThreadSleepTimer(500L);

        private final ExecutorService executors = new ThreadPoolExecutor(10, 10, 60000, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());

        private final ServiceThread reAssignHashSlotThread = new ServiceThread("defaultInnerMainSchedulerServiceThread") {
            @Override
            public void run() {
                while (true) {
                    try {
                        if (this.isStop()) {
                            return;
                        }
                        if (!DefaultSchedulerImpl.this.isMainScheduler.get()) {
                            // 防止内存泄漏/溢出
                            DefaultInnerMainScheduler.this.nettyEventQueue.clear();

                            reAssignHashSleepTimer.await();
                        } else {
                            int eventSize = nettyEventQueue.size();
                            int expectMaxReAssignEventSize = 10000;

                            if (eventSize >= expectMaxReAssignEventSize) {
                                reAssignHashSleepTimer.update();
                            } else {
                                reAssignHashSleepTimer.await();
                            }
                            List<NettyEvent> nettyEventList = new ArrayList<>();
                            DefaultInnerMainScheduler.this.nettyEventQueue.drainTo(nettyEventList, expectMaxReAssignEventSize);

                            if (CollectionUtils.isEmpty(nettyEventList)) {
                                continue;
                            }

                            for (NettyEvent nettyEvent : nettyEventList) {
                                String nettyEventIp = nettyEvent.getIp();
                                int nettyEventPort = nettyEvent.getPort();
                                NettyEventType nettyEventType = nettyEvent.getType();
                                String nettyEventAddr = NetworkUtil.splicingAddr(nettyEventIp, nettyEventPort);

                                switch (nettyEventType) {
                                    case EXCEPTION:
                                    case DISCONNECT: {
                                        schedulerAddrToHashSlotMap.remove(nettyEventAddr);
                                    }
                                    case CONNECT: {
                                        schedulerAddrToHashSlotMap.put(nettyEventAddr, new SchedulerHashSlotInfo(nettyEventIp, nettyEventPort));
                                    }
                                }
                            }

                            assignHashSlot();
                        }
                    } catch (Exception e) {
                        log.error("main scheduler reAssignHashSlotThread error ", e);
                    }
                }
            }
        };

        private Map<String/*addr*/, SchedulerHashSlotInfo> schedulerAddrToHashSlotMap = new HashMap<>();

        public DefaultInnerMainScheduler() {
            this.mainSchedulerRemoteServer = new NettyRemoteServer(schedulerConfig.getSelfMainSchedulerPort());
            this.mainSchedulerRemoteServer.registerEventManager(this);
        }

        @Override
        public synchronized void assignHashSlot() throws ExecutionException, InterruptedException {
            // hash 槽数量
            int schedulerHashSlotSize = SchedulerConstant.SCHEDULER_HASH_SLOT_SIZE;

            // 下发 hash 槽
            List<Integer> hashSlots = IntStream.rangeClosed(0, schedulerHashSlotSize - 1).boxed().collect(Collectors.toList());

            while (!doAssignHashSlot(hashSlots)) {
                Thread.sleep(5);
            }

            // todo 分配情况写到 db
        }

        /**
         * 下发 hash 槽
         */
        private boolean doAssignHashSlot(List<Integer> fullHashSlots) throws InterruptedException, ExecutionException {
            String selfMainSchedulerAddr = NetworkUtil.splicingAddr(schedulerConfig.getMainSchedulerIp(), schedulerConfig.getMainSchedulerPort());

            Set<String> schedulerAddrSet = this.schedulerAddrToHashSlotMap.keySet();
            List<String> schedulerAddrList = new ArrayList<>(schedulerAddrSet);

            // 主调度器 ip
            String mainSchedulerIp = DefaultSchedulerImpl.this.schedulerConfig.getSchedulerIp();
            // 主调度器端口号
            int mainSchedulerPort = DefaultSchedulerImpl.this.schedulerConfig.getSchedulerPort();

            // 调度器地址(ip + : + port) -> 为调度器分配的 hash 槽
            Map<String, List<Integer>> schedulerAddrToSlotMap = new HashMap<>();
            // 为调度器分配的 hash 槽 -> 调度器地址(ip + : + port)
            Map<Integer, String> slotToSchedulerAddrMap = new HashMap<>();
            // slot 分配信息
            List<SlotAssignInfo> slotAssignInfos = new ArrayList<>();

            int workSchedulerNum = schedulerAddrSet.size();

            // 分配 hash 槽
            for (int i = 0, schedulerIndex = 0;
                 i < fullHashSlots.size();
                 i++, schedulerIndex = (schedulerIndex + 1) % workSchedulerNum) {

                String schedulerAddr = schedulerAddrList.get(schedulerIndex);

                int hashSlot = fullHashSlots.get(i);
                SchedulerHashSlotInfo schedulerHashSlotInfo = this.schedulerAddrToHashSlotMap.get(schedulerAddr);

                schedulerHashSlotInfo.addHashSlot(hashSlot);

                schedulerAddrToSlotMap.computeIfAbsent(schedulerAddr, addr -> new ArrayList<>()).add(hashSlot);
                slotToSchedulerAddrMap.put(hashSlot, schedulerAddr);

                slotAssignInfos.add(
                        new SlotAssignInfo(
                                i,
                                schedulerHashSlotInfo.getSchedulerIp(),
                                schedulerHashSlotInfo.getSchedulerPort()
                        )
                );
            }

            // 下发哈希槽 todo 改为配置化
            long assignHashSlotAwaitTime = 5000L;

            Map<String/*addr*/, Future<SchedulerHashSlotInfo>> assignHashSlotFutureMap = new HashMap<>();
            CountDownLatch assignHashSlotCountDownLatch = new CountDownLatch(workSchedulerNum);

            this.schedulerAddrToHashSlotMap.forEach((schedulerAddr, schedulerHashSlotInfo) -> {
                Future<SchedulerHashSlotInfo> future = executors.submit(() -> {
                    try {
                        List<Integer> hashSlots = schedulerHashSlotInfo.getHashSlots();

                        SchedulerMetaData schedulerMetaData = new SchedulerMetaData(
                                schedulerAddrToSlotMap,
                                slotToSchedulerAddrMap,
                                hashSlots,
                                mainSchedulerIp,
                                mainSchedulerPort);

                        if (StringUtils.equals(schedulerAddr, selfMainSchedulerAddr)) {
                            DefaultSchedulerImpl.this.schedulerMetaData = schedulerMetaData;
                        } else {

                            AssignHashSlotsResponse assignHashSlotsResponse = this.mainSchedulerRemoteServer.send(
                                    schedulerAddr,
                                    new AssignHashSlotsRequest(schedulerMetaData),
                                    assignHashSlotAwaitTime,
                                    AssignHashSlotsResponse.class);

                            if (assignHashSlotsResponse == null) {
                                log.warn("assign hash slot response is null, schedulerHashSlotInfo = {}", schedulerHashSlotInfo);
                                schedulerHashSlotInfo.setAssignSuccess(false);
                            }
                        }

                        schedulerHashSlotInfo.setAssignSuccess(true);
                    } catch (Exception e) {
                        log.warn("assign hash slot error, schedulerHashSlotInfo = {}", schedulerHashSlotInfo, e);
                        this.mainSchedulerRemoteServer.disConnect(schedulerAddr);
                        schedulerHashSlotInfo.setAssignSuccess(false);
                    }

                    return schedulerHashSlotInfo;
                });
                assignHashSlotFutureMap.put(schedulerAddr, future);
            });

            assignHashSlotCountDownLatch.await(assignHashSlotAwaitTime, TimeUnit.MILLISECONDS);

            boolean assignHashSlotSuccess = true;
            // 避免阻塞下一次分配
            for (Map.Entry<String, Future<SchedulerHashSlotInfo>> entry : assignHashSlotFutureMap.entrySet()) {
                String schedulerAddr = entry.getKey();
                Future<SchedulerHashSlotInfo> assignHashSlotFuture = entry.getValue();

                boolean currentAssignSuccess = false;

                if (!assignHashSlotFuture.isDone()) {
                    assignHashSlotFuture.cancel(true);
                } else {
                    currentAssignSuccess = assignHashSlotFuture.get().isAssignSuccess();
                }

                assignHashSlotSuccess = assignHashSlotSuccess && currentAssignSuccess;

                if (!assignHashSlotSuccess) {
                    this.schedulerAddrToHashSlotMap.remove(schedulerAddr);
                }
            }

            // 分配成功
            if (assignHashSlotSuccess) {
                logAssignInfo();
                DefaultSchedulerImpl.this.slotAssignService.batchUpdateSlotAssignInfo(slotAssignInfos);
            }

            return assignHashSlotSuccess;
        }

        private void logAssignInfo() {
            StringBuilder assignResultBuilder = new StringBuilder("assign hash slot success, result is [");
            this.schedulerAddrToHashSlotMap
                    .values()
                    .forEach(assignInfo ->
                            assignResultBuilder.append(
                                    String.format("scheduler ip = %s, scheduler port = %s, slots = (%s)",
                                            assignInfo.getSchedulerIp(),
                                            assignInfo.getSchedulerPort(),
                                            assignInfo.getHashSlots()
                                    ))
                    );
            assignResultBuilder.append("]");

            log.info(assignResultBuilder.toString());
        }

        @Override
        public void start() throws InterruptedException {
            SchedulerConfig schedulerConfig = DefaultSchedulerImpl.this.schedulerConfig;

            // 当前调度器为主调度器时，是否进行调度工作
            boolean selfMainSchedulerEnabledWork = schedulerConfig.isSelfMainSchedulerEnabledWork();
            if (selfMainSchedulerEnabledWork) {
                String addr = NetworkUtil.splicingAddr(schedulerConfig.getSchedulerIp(), schedulerConfig.getSelfMainSchedulerPort());
                this.schedulerAddrToHashSlotMap.put(addr, new SchedulerHashSlotInfo(schedulerConfig.getSchedulerIp(), schedulerConfig.getSchedulerPort()));
            }

            this.reAssignHashSlotThread.start();
            this.mainSchedulerRemoteServer.start();
        }

        @Override
        public void shutdown() {
            this.mainSchedulerRemoteServer.shutdown();
            this.reAssignHashSlotThread.stop(true);
        }

        @Override
        public String getSubscriberName() {
            return "defaultInnerMainScheduler";
        }

        @Override
        public synchronized void connect(NettyEvent event) {
            // 有新节点加入
            this.nettyEventQueue.add(event);
        }

        @Override
        public synchronized void disconnect(NettyEvent event) {
            this.nettyEventQueue.add(event);
        }

        @Override
        public synchronized void exception(NettyEvent event) {
            this.nettyEventQueue.add(event);
        }
    }
}
